package org.xiaoyu.rpc.remoting.transport.cllient;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.xiaoyu.rpc.common.constant.RpcError;
import org.xiaoyu.rpc.common.exception.RpcException;
import org.xiaoyu.rpc.remoting.codec.Codec;
import org.xiaoyu.rpc.remoting.transport.MyDynamicLengthFrameDecoder;
import org.xiaoyu.rpc.remoting.transport.NettyCodecAdapter;
import org.xiaoyu.rpc.remoting.transport.PacketCodecHandler;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created By Have
 * 2021/7/5 16:50
 */
@Slf4j
public class ChannelProvider {
    private static NioEventLoopGroup workGroup;
    private static Bootstrap bootstrap;
    public static final Map<String, Channel> CONNECTED_CHANNEL = new ConcurrentHashMap<>();
    private static final int MAX_RETRY_COUNT = 5;

    static {
        workGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(workGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
//                pipeline.addLast(new MyDynamicLengthFrameDecoder(1024));
//                pipeline.addLast(PacketCodecHandler.INSTANCE);
                NettyCodecAdapter adapter = new NettyCodecAdapter(new Codec());
                pipeline.addLast("encoder", adapter.getEncoder());
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast(NettyClientHandler.INSTANCE);
            }
        });
    }

    public static Channel get(InetSocketAddress inetSocketAddress) {
        String hostString = inetSocketAddress.getHostString();
        int port = inetSocketAddress.getPort();
        Channel channel = CONNECTED_CHANNEL.get(hostString + ":" + port);
        if (channel != null) {
            return channel;
        }
        connect(bootstrap, inetSocketAddress);
        return CONNECTED_CHANNEL.get(hostString + ":" + port);
    }

    public static void remove(Channel channel) {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
        String hostString = inetSocketAddress.getHostString();
        int port = inetSocketAddress.getPort();
        CONNECTED_CHANNEL.remove(hostString + ":" + port);
    }

    private static void connect(Bootstrap bootstrap, InetSocketAddress inetSocketAddress) {
        connect(bootstrap, inetSocketAddress, MAX_RETRY_COUNT);
    }

    private static void connect(Bootstrap bootstrap, InetSocketAddress inetSocketAddress, int retryCount) {
        try {
            ChannelFuture channelFuture = bootstrap.connect(inetSocketAddress)
                    .addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (future.isSuccess()) {
                                log.info("客户端连接成功!");
                                return;
                            }
                            if (retryCount == 0) {
                                log.error("已达到最大重试连接次数,放弃连接");
                                throw new RpcException(RpcError.CLIENT_CONNECT_COUNT_REACH_MAX);
                            }

                            int order = (MAX_RETRY_COUNT - retryCount) + 1;
                            // 本次重连的间隔
                            int delay = 1 << order;
                            log.error("{}: 连接失败，第 {} 次重连……", new Date(), order);
                            bootstrap.config().group().schedule(() -> connect(bootstrap, inetSocketAddress, retryCount - 1), delay, TimeUnit
                                    .SECONDS);
                        }
                    }).sync();
            Channel channel = channelFuture.sync().channel();

            InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();
            CONNECTED_CHANNEL.put(socketAddress.getHostString() + ":" + socketAddress.getPort(), channel);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}
